[api][runtime][python] Add minimal Event lineage#923
Conversation
github-actions
Bot
added
doc-label-missing
|
One ActionStateStore replay semantic seems worth making explicit before merging. When a completed action result is reused, the runtime returns the stored output Events and rebinds their lineage to the current triggering Event while preserving the output Event IDs. The resulting behavior is: I think this is a reasonable semantic, but it is currently implicit. Could we document it and add a test that asserts |
| warnings: list[dict[str, str]] = [] | ||
| nodes: dict[str, dict[str, Any]] = {} | ||
| for event_id, matching_records in records_by_id.items(): | ||
| if len(matching_records) > 1: |
There was a problem hiding this comment.
ActionStateStore reuse can legitimately log the same Event ID again, possibly with different upstream lineage. Treating every repeated ID as invalid and dropping all matching records loses valid replay branches. Could reconstruction model this as a DAG instead: deduplicate repeated (event ID, lineage edge) observations, retain distinct lineage edges for the same Event ID, and report an ID conflict only when the Event type or content is inconsistent?
| } | ||
|
|
||
| /** Copies framework-managed metadata when reconstructing a typed Event. */ | ||
| protected void copyFrameworkMetadataFrom(Event event) { |
There was a problem hiding this comment.
Typed reconstruction represents the same Event occurrence, but preserving that occurrence currently depends on every custom fromEvent implementation remembering both to pass the ID separately and to call this helper. Could the framework own a reconstruction template/factory so a custom subtype only maps its payload? Otherwise an omitted helper call silently loses sourceTimestamp and lineage.
| if name != "id": | ||
| object.__setattr__(self, "id", self._generate_content_based_id()) | ||
|
|
||
| def with_framework_metadata_from(self, source: "Event") -> Self: |
There was a problem hiding this comment.
The Java and Python helpers currently have different contracts: Python returns a copy and includes the Event ID, while Java mutates the current object and does not copy the ID. Also, “framework metadata” describes ownership rather than the Event-domain meaning. Could the names and contracts be aligned around preserving/reconstructing the same Event occurrence?
| } | ||
| } | ||
|
|
||
| private static void setOutputEventLineage(ActionTask actionTask, List<Event> outputEvents) { |
There was a problem hiding this comment.
This rule describes how a concrete Action invocation finalizes its output Events, rather than Operator scheduling. Could it be owned by ActionTask, which already holds the triggering Event and Action? That would keep the Operator independent of lineage field details and provide one place to reject outputEvent.id == triggerEvent.id; currently ctx.sendEvent(event) can create a self-loop and mutate the triggering Event shared by sibling ActionTasks.
| # Set fallback if not provided in kwargs | ||
| if "fallback" not in kwargs: | ||
| kwargs["fallback"] = self.__serialize_unknown | ||
| if "by_alias" not in kwargs: |
There was a problem hiding this comment.
Defaulting by_alias to True here changes serialization for every alias declared by a custom Event subclass, not only the two lineage fields. Could the camelCase mapping be scoped to the lineage fields, or should this broader compatibility change be documented and covered by a custom-alias test?
2 participants
Linked issue: Closes #841
Related discussions:
Purpose of change
The Event Log currently records individual Events but does not preserve enough information to reconstruct the direct causal path between an Event, the Action it triggered, and the Events emitted by that Action.
This PR implements the minimal Event-lineage phase discussed in #710.
It adds two framework-managed fields to each Event emitted by an Action:
upstreamEventId: the ID of the Event consumed by the emitting Action;upstreamActionName: the name of the Action that emitted the current Event.Together with the existing per-occurrence Event ID, these fields are sufficient to reconstruct an InputEvent-rooted Event-Action-Event causal tree from the available Event Log records.
This PR deliberately does not implement the complete execution-tracing model proposed in #900. Run identity, concrete Action invocation identity, execution hierarchy, lifecycle, retry/replay semantics, ordering, and failure details remain part of that separate design.
A real Runtime execution was verified with the following causal path:
The File Event Log contained three physical Event records:
upstreamEventIdupstreamActionName_input_eventMiddleEventaction1_output_eventaction2Running
tools/reconstruct_trace_tree.pyagainst that Runtime log produced:The JSON reconstruction contained one root, all three Event nodes, the two expected virtual Action edges, and no warnings.
Tests
Add e2e test to review Trace Tree
python/flink_agents/e2e_tests/e2e_tests_integration/python_event_logging_test.pyAPI
This PR changes the default Python Event ID generation strategy from content-derived IDs to UUIDv4.
Documentation
doc-neededdoc-not-neededdoc-included